package com.atguigu.edu.realtime.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @Author zhangsan
 * @Date 2022/10/18 23:08
 * @Description //TODO 维度查询的工具类
 */
public class DimUtil {
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }


    public static JSONObject getDimInfo(String tableName, Tuple2<String,String> ... columnNameAndValues){

        //拼接从redis中查询维度的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase() + ":");


        //拼接查询语句
        //String selectSql = "select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where id ='" + id +"'";
        StringBuilder selectSql = new StringBuilder("select * from " + EduConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        //System.out.println(selectSql);
        //对所有的查询条件遍历
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            //System.out.println("columnName" + columnName);
            String columnValue = columnNameAndValue.f1;
            //System.out.println("columnValue" + columnValue);
            selectSql.append(columnName + "='" + columnValue + "'");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length-1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }
        //先从redis中获取维度数据
        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            dimInfoStr  = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("~~从Redis中查询维度数据发生了异常~~");
        }

        //判断是否缓存命中
        if (StringUtils.isNotEmpty(dimInfoStr)){
            //缓存命中 直接将字符穿转换为json对象 作为查询的维度返回
            dimJsonObj = JSON.parseObject(dimInfoStr);

        }else {
            //缓存未命中   发送请求到phoenix表中查询维度数据，并将查询的维度放到Redis中缓存起来
            System.out.println("从phoenix表中查询维度数据的SQL:" + selectSql);

            //底层还是调用phonenixUtil工具类中的queryList方法
            List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            //因为根据主键查询维度数据，所以如果查到维度数据，集合中也会只有一条数据
            if (dimInfoJsonObjList!=null && dimInfoJsonObjList.size()>0){
                dimJsonObj = dimInfoJsonObjList.get(0);
                //将查询的维度放到Redis中缓存起来
                if (jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimJsonObj.toJSONString());
                }

            }else {
                System.out.println("在维度表中没有找到对应的数据");
            }
        }

        //释放资源
        if (jedis !=null){
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }

        return dimJsonObj;
    }



    //根据维度的主键 到维度表中查询维度数据 没有任何优化
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String,String> ... columnNameAndValues){


        //拼接查询语句
        //String selectSql = "select * from " + EduConfig.PHOENIX_SCHEMA + "." + tableName + " where id ='" + id +"'";
        StringBuilder selectSql = new StringBuilder("select * from " + EduConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        //对所有的查询条件遍历
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if (i < columnNameAndValues.length-1){
                selectSql.append(" and ");
            }
        }
        System.out.println("从phoenix表中查询维度数据的SQL:" + selectSql);

        //底层还是调用phonenixUtil工具类中的queryList方法
        List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        //因为根据主键查询维度数据，所以如果查到维度数据，集合中也会只有一条数据
        JSONObject dimJsonObj = null;
        if (dimInfoJsonObjList!=null && dimInfoJsonObjList.size()>0){
            dimJsonObj = dimInfoJsonObjList.get(0);
        }else {
            System.out.println("在维度表中没有找到对应的数据");
        }
        return dimJsonObj;
    }


    public static void deleteCached(String tableName, String id){
        String redisKey =  "dim:" + tableName.toLowerCase() + ":"+id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis!=null){
                System.out.println("~~删除缓存后，关闭Jedis连接~~");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) {
        //System.out.println(getDimInfoNoCache("dim_base_source", Tuple2.of("id","1")));
        System.out.println(getDimInfo("dim_base_source", "1"));
    }
}
